package jupitermouse.site.example.scala.window

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer

object KafkaConnectorProducerApp {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //从socket接受数据，通过flink将数据sink到Kafka
    val data = env.socketTextStream("localhost", 9999)


 /*   val kafkaSink = new FlinkKafkaProducer[String](
      "192.168.81.130:9092",         // broker list
      "flink",               // target topic
      new SimpleStringSchema)
*/
 val kafkaSink = new FlinkKafkaProducer[String](
     "192.168.81.130:9092",         // broker list
     "flink",               // target topic
     new SimpleStringSchema)

    //checkpoint常用参数设置
    env.enableCheckpointing(4000)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.setCheckpointTimeout(10000)
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

    FlinkKafkaProducer.Semantic.EXACTLY_ONCE

    kafkaSink.setWriteTimestampToKafka(true)

    data.addSink(kafkaSink)

    env.execute("KafkaConnectorProducerApp")
  }
}
